package com.heima.article.stream;

import com.alibaba.fastjson.JSON;
import com.heima.article.config.stream.KafkaStreamListener;
import com.heima.common.constants.message.HotArticleConstants;
import com.heima.model.article.mess.ArticleVisitStreamMess;
import com.heima.model.article.mess.UpdateArticleMess;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.streams.KeyValue;
import org.apache.kafka.streams.kstream.KStream;
import org.apache.kafka.streams.kstream.TimeWindows;
import org.springframework.stereotype.Component;

import java.time.Duration;

@Component//被Spring识别
@Slf4j
public class HotArticleStreamHandler implements KafkaStreamListener<KStream<String,String>> {




    @Override
    public String listenerTopic() {//监听的主题
        return HotArticleConstants.HOTARTICLE_SCORE_INPUT_TOPIC;
    }

    @Override
    public String sendTopic() {
        return HotArticleConstants.HOTARTICLE_INCR_HANDLE_OUPUT_TOPIC;
    }

    @Override
    public KStream<String, String> getService(KStream<String, String> stream) {
        return stream.map((key,value)->{
            log.info("流式处理已被触发,收到的信息内容为====>"+value);
            //得到是json字符串   转化为  json对象
            UpdateArticleMess updateArticleMess = JSON.parseObject(value, UpdateArticleMess.class);//指定转化类型
            String newKey =updateArticleMess.getArticleId()+":"+updateArticleMess.getType().name();//得到类型的字符串
            String newValue=String.valueOf(updateArticleMess.getAdd());
            return new KeyValue<>(newKey,newValue);
        }).groupByKey()//分组后要计算
                .windowedBy(TimeWindows.of(Duration.ofSeconds(10)))
                .aggregate(()->"0",(k,v,preR)->{
                    return String.valueOf(Integer.valueOf(preR)+Integer.valueOf(v));
                }).toStream()
                        .map((key,value)->{
                            String articleIdAndType = key.key();//key:变成了一个对象  (原来的分组的key  articleId:type  时间戳)
                            Integer result = Integer.valueOf(value);
                            ArticleVisitStreamMess streamMess = formatObj(articleIdAndType,result);
                            return new KeyValue<>("",JSON.toJSONString(streamMess));
                        });//格式化  key:value
    }


    private ArticleVisitStreamMess formatObj(String articleIdAndType, Integer result) {
        //用一个新的类来接收参数
        ArticleVisitStreamMess  streamMess= new ArticleVisitStreamMess();
        String[] split = articleIdAndType.split(":");
        String articleId=split[0];
        String typeName=split[1];
        streamMess.setArticleId(Long.valueOf(articleId));
        if (typeName.equals(UpdateArticleMess.UpdateArticleType.VIEWS.name())){//字符串相等
            streamMess.setView(result);
        }
        if(typeName.equals(UpdateArticleMess.UpdateArticleType.LIKES.name())){
            streamMess.setLike(result);
        }
        if(typeName.equals(UpdateArticleMess.UpdateArticleType.COMMENT.name())){
            streamMess.setComment(result);
        }
        if(typeName.equals(UpdateArticleMess.UpdateArticleType.COLLECTION.name())){
            streamMess.setCollect(result);
        }
        return streamMess;
    }
}
